package com.lcn.core.service.aop;

import com.lcn.comm.domain.TxGroup;
import com.lcn.comm.domain.TxTransactionInfo;
import com.lcn.comm.domain.TxTransactionLocal;
import com.lcn.comm.framework.task.ConditionUtils;
import com.lcn.comm.framework.task.Task;
import com.lcn.comm.framework.task.TaskGroupManager;
import com.lcn.comm.framework.task.TxTask;
import com.lcn.comm.framework.thread.HookRunnable;
import com.lcn.comm.service.aop.TransactionService;
import com.lcn.comm.service.transaction.TransactionGroupService;
import com.lcn.comm.util.PoolUtil;
import com.lcn.comm.util.RequestIdGenerator;
import com.weibo.api.motan.config.springsupport.BasicServiceConfigBean;
import com.weibo.api.motan.config.springsupport.annotation.MotanReferer;
import org.aspectj.lang.ProceedingJoinPoint;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * <p>分布式事务启动开始时的业务处理</p>
 *
 * @author 张峰 zfvip_it@163.com
 * @create: 下午9:19
 */
@Service("startTransactionService")
public class StartTransactionServiceImpl implements TransactionService {

    @MotanReferer
    private TransactionGroupService transactionGroupService;

    @Resource
    private BasicServiceConfigBean basicServiceConfigBean;


    /**
     * 事务逻辑处理
     *
     * @param point
     * @param info
     * @return
     * @throws Throwable
     */
    public Object execute(ProceedingJoinPoint point, TxTransactionInfo info) throws Throwable {
        //创建事务组
        TxGroup txGroup = transactionGroupService.createTransactionGroup(null);
        if (txGroup != null) {
            //将事务组信息缓存到当前线程
            TxTransactionLocal txTransactionLocal = cacheTxTransactionLocal(txGroup);
            //用于标记是否正常执行
            int state = 0;
            try {
                Object obj = point.proceed();
                state = 1;
                return obj;
            } catch (Throwable e) {
                //事务回滚处理
                throw e;
            } finally {
                handler(txTransactionLocal, state);
            }
        } else {
            throw new RuntimeException("事务组创建失败");
        }
    }

    /**
     * 事务发起方业务完成处理
     *
     * @param txTransactionLocal
     * @param state
     */
    private void handler(TxTransactionLocal txTransactionLocal, int state) {
        if (!txTransactionLocal.isHasOnlyNode()) {
            //创建任务，用于保证其他业务都已经执行
            Task task = ConditionUtils.getInstance().createTask(RequestIdGenerator.getRequestId());
            //通知关闭事务组
            notifyCloseTransactionGroup(task, txTransactionLocal, state);
            //等待其他业务处理结束
            task.waitTask();
            //删除task
            task.remove();
        } else {
            //控制本地事务的数据提交
            signalTask(txTransactionLocal, state);
            //异步通知关闭事务组
            synCloseTransactionGroup(txTransactionLocal, state);
        }
        TxTransactionLocal.setCurrent(null);
    }

    /**
     * 异步通知关闭事务组
     *
     * @param txTransactionLocal
     * @param state
     */
    private void synCloseTransactionGroup(final TxTransactionLocal txTransactionLocal, final int state) {
        PoolUtil.getPool().execute(new HookRunnable() {
            @Override
            public void run0() {
                transactionGroupService.closeTransactionGroup(txTransactionLocal.getGroupId(), state);
            }
        });
    }

    /**
     * 异步通知关闭事务组
     *
     * @param task
     * @param txTransactionLocal
     * @param state
     */
    private void notifyCloseTransactionGroup(final Task task, final TxTransactionLocal txTransactionLocal, final int state) {
        PoolUtil.getPool().execute(new HookRunnable() {
            public void run0() {
                while (!task.isAwait() && !Thread.currentThread().interrupted()) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(1L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //关闭事务组
                int res = transactionGroupService.closeTransactionGroup(txTransactionLocal.getGroupId(), state);
                int lastState = res == 0 ? 0 : state;
                //控制本地事务的数据提交
                signalTask(txTransactionLocal, lastState);
                task.setState(lastState);
                task.signalTask();
            }
        });
    }

    /**
     * 唤醒task
     *
     * @param txTransactionLocal
     * @param state
     */
    private void signalTask(TxTransactionLocal txTransactionLocal, int state) {
        //控制本地事务的数据提交
        TxTask txTask = TaskGroupManager.getInstance().getTask(txTransactionLocal.getGroupId(), txTransactionLocal.getRequestId());
        if (txTask != null) {
            txTask.setState(state);
            //唤醒task
            txTask.signalTask();
            while (!txTask.isRemove()) {
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 将事务组信息缓存到当前线程
     *
     * @param txGroup
     */
    private TxTransactionLocal cacheTxTransactionLocal(TxGroup txGroup) {
        if (txGroup != null) {
            TxTransactionLocal txTransactionLocal = new TxTransactionLocal();
            //事务组ID
            txTransactionLocal.setGroupId(txGroup.getGroupId());
            //请求id
            txTransactionLocal.setRequestId(RequestIdGenerator.getRequestId());
            //是否发起方模块
            txTransactionLocal.setHasStart(true);
            //请求超时时间
            txTransactionLocal.setRequestTimeout(basicServiceConfigBean.getRequestTimeout());
            //ip端口
            txTransactionLocal.setIpPort(basicServiceConfigBean.getExport());
            TxTransactionLocal.setCurrent(txTransactionLocal);
            return txTransactionLocal;
        }
        return null;
    }
}
